//! # Mock框架
//!
//! 提供消息总线组件的Mock实现，用于单元测试

use crate::bus::{MessageBus, Publisher, Subscriber};
use crate::message::{Message, MessageMetadata, MessagePriority, MessageType, PersistenceLevel};
use crate::error::{MessageBusError, MessageBusResult};
use async_trait::async_trait;
use std::collections::{HashMap, VecDeque};
use std::sync::{Arc, Mutex};
use std::time::Duration;
use tokio::sync::RwLock;
use tokio::time::sleep;
use uuid::Uuid;
use chrono::Utc;

/// Mock行为配置
#[derive(Debug, Clone)]
pub struct MockBehavior {
    /// 是否应该失败
    pub should_fail: bool,
    /// 失败错误消息
    pub error_message: Option<String>,
    /// 延迟时间（毫秒）
    pub delay_ms: u64,
    /// 是否应该超时
    pub should_timeout: bool,
    /// 限制调用次数
    pub call_limit: Option<usize>,
    /// 当前调用次数
    pub call_count: usize,
}

impl Default for MockBehavior {
    fn default() -> Self {
        Self {
            should_fail: false,
            error_message: None,
            delay_ms: 0,
            should_timeout: false,
            call_limit: None,
            call_count: 0,
        }
    }
}

impl MockBehavior {
    pub fn new() -> Self {
        Self::default()
    }

    pub fn with_failure(mut self, error_message: impl Into<String>) -> Self {
        self.should_fail = true;
        self.error_message = Some(error_message.into());
        self
    }

    pub fn with_delay(mut self, delay_ms: u64) -> Self {
        self.delay_ms = delay_ms;
        self
    }

    pub fn with_timeout(mut self) -> Self {
        self.should_timeout = true;
        self
    }

    pub fn with_call_limit(mut self, limit: usize) -> Self {
        self.call_limit = Some(limit);
        self
    }

    pub async fn execute(&mut self) -> MessageBusResult<()> {
        self.call_count += 1;

        // 检查调用次数限制
        if let Some(limit) = self.call_limit {
            if self.call_count > limit {
                return Err(MessageBusError::internal_error("调用次数超出限制"));
            }
        }

        // 模拟延迟
        if self.delay_ms > 0 {
            sleep(Duration::from_millis(self.delay_ms)).await;
        }

        // 模拟超时
        if self.should_timeout {
            sleep(Duration::from_secs(10)).await; // 长时间延迟模拟超时
        }

        // 模拟失败
        if self.should_fail {
            let error_msg = self.error_message.clone()
                .unwrap_or_else(|| "Mock failure".to_string());
            return Err(MessageBusError::internal_error(error_msg));
        }

        Ok(())
    }
}

/// Mock消息总线状态
#[derive(Debug)]
pub struct MockMessageBusState {
    /// 发布的消息
    pub published_messages: Vec<Message>,
    /// 订阅者映射
    pub subscribers: HashMap<String, Vec<MockSubscriber>>,
    /// 主题统计
    pub topic_stats: HashMap<String, usize>,
    /// 是否已关闭
    pub is_closed: bool,
}

impl Default for MockMessageBusState {
    fn default() -> Self {
        Self {
            published_messages: Vec::new(),
            subscribers: HashMap::new(),
            topic_stats: HashMap::new(),
            is_closed: false,
        }
    }
}

/// Mock消息总线实现
#[derive(Clone)]
pub struct MockMessageBus {
    state: Arc<RwLock<MockMessageBusState>>,
    behaviors: Arc<RwLock<HashMap<String, MockBehavior>>>,
}

impl MockMessageBus {
    pub fn new() -> Self {
        Self {
            state: Arc::new(RwLock::new(MockMessageBusState::default())),
            behaviors: Arc::new(RwLock::new(HashMap::new())),
        }
    }

    /// 设置方法行为
    pub async fn set_behavior(&self, method: &str, behavior: MockBehavior) {
        let mut behaviors = self.behaviors.write().await;
        behaviors.insert(method.to_string(), behavior);
    }

    /// 获取并执行行为
    async fn execute_behavior(&self, method: &str) -> MessageBusResult<()> {
        let mut behaviors = self.behaviors.write().await;
        if let Some(behavior) = behaviors.get_mut(method) {
            behavior.execute().await
        } else {
            Ok(())
        }
    }

    /// 获取发布的消息
    pub async fn get_published_messages(&self) -> Vec<Message> {
        self.state.read().await.published_messages.clone()
    }

    /// 获取指定主题的发布消息
    pub async fn get_published_messages_for_topic(&self, topic: &str) -> Vec<Message> {
        let state = self.state.read().await;
        state.published_messages
            .iter()
            .filter(|msg| msg.topic() == topic)
            .cloned()
            .collect()
    }

    /// 清空发布的消息
    pub async fn clear_published_messages(&self) {
        let mut state = self.state.write().await;
        state.published_messages.clear();
    }

    /// 检查是否已关闭
    pub async fn is_closed(&self) -> bool {
        self.state.read().await.is_closed
    }

    /// 获取订阅者数量
    pub async fn get_subscriber_count(&self, topic: &str) -> usize {
        let state = self.state.read().await;
        state.subscribers.get(topic).map_or(0, |subs| subs.len())
    }
}

#[async_trait]
impl MessageBus for MockMessageBus {
    async fn publish(&self, message: Message) -> MessageBusResult<()> {
        self.execute_behavior("publish").await?;

        let mut state = self.state.write().await;
        if state.is_closed {
            return Err(MessageBusError::internal_error("消息总线已关闭"));
        }

        // 记录消息
        state.published_messages.push(message.clone());
        
        // 更新统计
        let count = state.topic_stats.entry(message.topic().to_string()).or_insert(0);
        *count += 1;

        // 分发给订阅者
        let topic = message.topic();
        if let Some(subscribers) = state.subscribers.get_mut(topic) {
            for subscriber in subscribers.iter_mut() {
                subscriber.add_message(message.clone()).await;
            }
        }

        // 处理通配符订阅
        for (pattern, subscribers) in state.subscribers.iter_mut() {
            if pattern.contains('*') && Self::matches_pattern(topic, pattern) {
                for subscriber in subscribers.iter_mut() {
                    subscriber.add_message(message.clone()).await;
                }
            }
        }

        Ok(())
    }

    async fn subscribe(&self, pattern: &str) -> MessageBusResult<Box<dyn Subscriber>> {
        self.execute_behavior("subscribe").await?;

        let subscriber = MockSubscriber::new(pattern);
        let boxed_subscriber = Box::new(subscriber.clone()) as Box<dyn Subscriber>;

        let mut state = self.state.write().await;
        state.subscribers
            .entry(pattern.to_string())
            .or_insert_with(Vec::new)
            .push(subscriber);

        Ok(boxed_subscriber)
    }

    async fn unsubscribe(&self, subscription_id: &str) -> MessageBusResult<()> {
        self.execute_behavior("unsubscribe").await?;

        let mut state = self.state.write().await;
        for subscribers in state.subscribers.values_mut() {
            subscribers.retain(|sub| sub.subscription_id() != subscription_id);
        }

        Ok(())
    }

    async fn subscriber_count(&self, topic: &str) -> MessageBusResult<usize> {
        self.execute_behavior("subscriber_count").await?;

        let state = self.state.read().await;
        Ok(state.subscribers.get(topic).map_or(0, |subs| subs.len()))
    }

    async fn list_topics(&self) -> MessageBusResult<Vec<String>> {
        self.execute_behavior("list_topics").await?;

        let state = self.state.read().await;
        Ok(state.topic_stats.keys().cloned().collect())
    }

    async fn has_subscribers(&self, topic: &str) -> MessageBusResult<bool> {
        self.execute_behavior("has_subscribers").await?;

        let count = self.subscriber_count(topic).await?;
        Ok(count > 0)
    }

    async fn close(&self) -> MessageBusResult<()> {
        self.execute_behavior("close").await?;

        let mut state = self.state.write().await;
        state.is_closed = true;

        // 关闭所有订阅者
        for subscribers in state.subscribers.values_mut() {
            for subscriber in subscribers.iter_mut() {
                let _ = subscriber.close().await;
            }
        }

        Ok(())
    }

    async fn stats(&self) -> MessageBusResult<crate::bus::MessageBusStats> {
        self.execute_behavior("stats").await?;

        let state = self.state.read().await;
        Ok(crate::bus::MessageBusStats {
            total_published: state.published_messages.len() as u64,
            total_received: 0, // Mock不跟踪接收
            total_failed: 0,
            active_subscribers: state.subscribers.values().map(|v| v.len()).sum(),
            active_topics: state.topic_stats.len(),
            queue_length: 0,
            uptime_seconds: 0,
            messages_per_second: 0.0,
            average_message_size: 0,
            memory_usage: 0,
        })
    }
}

impl MockMessageBus {
    /// 检查主题是否匹配模式
    fn matches_pattern(topic: &str, pattern: &str) -> bool {
        if pattern == "*" || pattern == "**" {
            return true;
        }

        let topic_parts: Vec<&str> = topic.split('.').collect();
        let pattern_parts: Vec<&str> = pattern.split('.').collect();

        Self::match_parts(&topic_parts, &pattern_parts)
    }

    fn match_parts(topic_parts: &[&str], pattern_parts: &[&str]) -> bool {
        let mut topic_idx = 0;
        let mut pattern_idx = 0;

        while topic_idx < topic_parts.len() && pattern_idx < pattern_parts.len() {
            match pattern_parts[pattern_idx] {
                "*" => {
                    // 单级通配符
                    topic_idx += 1;
                    pattern_idx += 1;
                }
                "**" => {
                    // 多级通配符
                    if pattern_idx == pattern_parts.len() - 1 {
                        return true; // ** 在最后，匹配剩余所有
                    }
                    pattern_idx += 1;
                    // 寻找下一个匹配的部分
                    while topic_idx < topic_parts.len() {
                        if Self::match_parts(&topic_parts[topic_idx..], &pattern_parts[pattern_idx..]) {
                            return true;
                        }
                        topic_idx += 1;
                    }
                    return false;
                }
                part => {
                    if topic_parts[topic_idx] != part {
                        return false;
                    }
                    topic_idx += 1;
                    pattern_idx += 1;
                }
            }
        }

        topic_idx == topic_parts.len() && pattern_idx == pattern_parts.len()
    }
}

/// Mock订阅者状态
#[derive(Debug)]
pub struct MockSubscriberState {
    pub subscription_id: String,
    pub pattern: String,
    pub messages: VecDeque<Message>,
    pub is_active: bool,
    pub acknowledged_messages: Vec<Uuid>,
    pub nacked_messages: Vec<(Uuid, bool)>, // (message_id, requeue)
}

/// Mock订阅者实现
#[derive(Clone)]
pub struct MockSubscriber {
    state: Arc<Mutex<MockSubscriberState>>,
    behaviors: Arc<RwLock<HashMap<String, MockBehavior>>>,
}

impl MockSubscriber {
    pub fn new(pattern: &str) -> Self {
        let subscription_id = Uuid::new_v4().to_string();
        Self {
            state: Arc::new(Mutex::new(MockSubscriberState {
                subscription_id,
                pattern: pattern.to_string(),
                messages: VecDeque::new(),
                is_active: true,
                acknowledged_messages: Vec::new(),
                nacked_messages: Vec::new(),
            })),
            behaviors: Arc::new(RwLock::new(HashMap::new())),
        }
    }

    /// 设置方法行为
    pub async fn set_behavior(&self, method: &str, behavior: MockBehavior) {
        let mut behaviors = self.behaviors.write().await;
        behaviors.insert(method.to_string(), behavior);
    }

    /// 获取并执行行为
    async fn execute_behavior(&self, method: &str) -> MessageBusResult<()> {
        let mut behaviors = self.behaviors.write().await;
        if let Some(behavior) = behaviors.get_mut(method) {
            behavior.execute().await
        } else {
            Ok(())
        }
    }

    /// 添加消息到队列
    pub async fn add_message(&mut self, message: Message) {
        let mut state = self.state.lock().unwrap();
        if state.is_active {
            state.messages.push_back(message);
        }
    }

    /// 获取队列中的消息数量
    pub fn message_count(&self) -> usize {
        self.state.lock().unwrap().messages.len()
    }

    /// 获取已确认的消息ID列表
    pub fn get_acknowledged_messages(&self) -> Vec<Uuid> {
        self.state.lock().unwrap().acknowledged_messages.clone()
    }

    /// 获取已拒绝的消息ID列表
    pub fn get_nacked_messages(&self) -> Vec<(Uuid, bool)> {
        self.state.lock().unwrap().nacked_messages.clone()
    }

    /// 清空消息队列
    pub fn clear_messages(&self) {
        let mut state = self.state.lock().unwrap();
        state.messages.clear();
    }
}

#[async_trait]
impl Subscriber for MockSubscriber {
    async fn try_receive(&self) -> MessageBusResult<Option<Message>> {
        self.execute_behavior("try_receive").await?;

        let mut state = self.state.lock().unwrap();
        if !state.is_active {
            return Err(MessageBusError::internal_error("订阅者已关闭"));
        }

        Ok(state.messages.pop_front())
    }

    async fn receive(&self) -> MessageBusResult<Option<Message>> {
        self.execute_behavior("receive").await?;

        // 对于Mock，我们简单地调用try_receive，实际实现会阻塞等待
        self.try_receive().await
    }

    async fn receive_timeout(&self, timeout_ms: u64) -> MessageBusResult<Option<Message>> {
        self.execute_behavior("receive_timeout").await?;

        // 模拟超时等待
        let start = std::time::Instant::now();
        let timeout_duration = Duration::from_millis(timeout_ms);

        while start.elapsed() < timeout_duration {
            if let Some(message) = self.try_receive().await? {
                return Ok(Some(message));
            }
            sleep(Duration::from_millis(10)).await;
        }

        Ok(None) // 超时
    }

    async fn ack(&self, message: &Message) -> MessageBusResult<()> {
        self.execute_behavior("ack").await?;

        let mut state = self.state.lock().unwrap();
        state.acknowledged_messages.push(message.id());
        Ok(())
    }

    async fn nack(&self, message: &Message, requeue: bool) -> MessageBusResult<()> {
        self.execute_behavior("nack").await?;

        let mut state = self.state.lock().unwrap();
        state.nacked_messages.push((message.id(), requeue));

        // 如果重新入队，将消息放回队列前端
        if requeue {
            state.messages.push_front(message.clone());
        }

        Ok(())
    }

    fn subscription_id(&self) -> &str {
        // 注意：这里需要返回引用，但我们的状态在Mutex中
        // 在实际测试中，我们可以通过其他方式获取ID
        "mock_subscription_id"
    }

    fn pattern(&self) -> &str {
        // 同样的问题，在实际测试中需要特别处理
        "mock_pattern"
    }

    fn is_active(&self) -> bool {
        self.state.lock().unwrap().is_active
    }

    async fn close(&self) -> MessageBusResult<()> {
        self.execute_behavior("close").await?;

        let mut state = self.state.lock().unwrap();
        state.is_active = false;
        state.messages.clear();
        Ok(())
    }

    async fn stats(&self) -> MessageBusResult<crate::subscriber::SubscriberStats> {
        self.execute_behavior("stats").await?;

        let state = self.state.lock().unwrap();
        Ok(crate::subscriber::SubscriberStats {
            total_received: state.acknowledged_messages.len() as u64,
            total_acked: state.acknowledged_messages.len() as u64,
            total_nacked: state.nacked_messages.len() as u64,
            queue_length: state.messages.len(),
            average_processing_time_ms: 0.0,
            last_received_at: None,
            subscribed_at: Utc::now(),
        })
    }
}

/// Mock消息构建器
pub struct MockMessageBuilder {
    topic: String,
    payload: serde_json::Value,
    metadata: MessageMetadata,
}

impl MockMessageBuilder {
    pub fn new(topic: &str) -> Self {
        Self {
            topic: topic.to_string(),
            payload: serde_json::Value::Null,
            metadata: MessageMetadata::new(topic),
        }
    }

    pub fn with_payload(mut self, payload: serde_json::Value) -> Self {
        self.payload = payload;
        self
    }

    pub fn with_message_type(mut self, message_type: MessageType) -> Self {
        self.metadata.attributes.message_type = message_type;
        self
    }

    pub fn with_priority(mut self, priority: MessagePriority) -> Self {
        self.metadata.attributes.priority = priority;
        self
    }

    pub fn with_persistence(mut self, persistence: PersistenceLevel) -> Self {
        self.metadata.attributes.persistence = persistence;
        self
    }

    pub fn with_correlation_id(mut self, correlation_id: Uuid) -> Self {
        self.metadata.correlation_id = Some(correlation_id);
        self
    }

    pub fn with_reply_to(mut self, reply_to: &str) -> Self {
        self.metadata.reply_to = Some(reply_to.to_string());
        self
    }

    pub fn with_max_retries(mut self, max_retries: u32) -> Self {
        self.metadata.attributes.max_retries = max_retries;
        self
    }

    pub fn build(self) -> Message {
        Message::with_metadata(self.payload, self.metadata)
    }
}

/// 测试辅助函数
pub mod test_helpers {
    use super::*;
    use serde_json::json;

    /// 创建测试消息
    pub fn create_test_message(topic: &str, payload: serde_json::Value) -> Message {
        MockMessageBuilder::new(topic)
            .with_payload(payload)
            .build()
    }

    /// 创建事件消息
    pub fn create_event_message(topic: &str, payload: serde_json::Value) -> Message {
        MockMessageBuilder::new(topic)
            .with_payload(payload)
            .with_message_type(MessageType::Event)
            .build()
    }

    /// 创建命令消息
    pub fn create_command_message(topic: &str, payload: serde_json::Value) -> Message {
        MockMessageBuilder::new(topic)
            .with_payload(payload)
            .with_message_type(MessageType::Command)
            .with_priority(MessagePriority::High)
            .build()
    }

    /// 创建带关联ID的消息
    pub fn create_correlated_message(topic: &str, payload: serde_json::Value, correlation_id: Uuid) -> Message {
        MockMessageBuilder::new(topic)
            .with_payload(payload)
            .with_correlation_id(correlation_id)
            .build()
    }

    /// 创建带回复地址的消息
    pub fn create_request_message(topic: &str, payload: serde_json::Value, reply_to: &str) -> Message {
        MockMessageBuilder::new(topic)
            .with_payload(payload)
            .with_message_type(MessageType::Query)
            .with_reply_to(reply_to)
            .build()
    }

    /// 创建批量测试消息
    pub fn create_batch_messages(topic: &str, count: usize) -> Vec<Message> {
        (0..count)
            .map(|i| {
                create_test_message(
                    topic,
                    json!({
                        "id": i,
                        "content": format!("message_{}", i),
                        "timestamp": Utc::now().to_rfc3339()
                    })
                )
            })
            .collect()
    }

    /// 等待条件满足
    pub async fn wait_for_condition<F>(condition: F, timeout_ms: u64) -> bool
    where
        F: Fn() -> bool,
    {
        let start = std::time::Instant::now();
        let timeout_duration = Duration::from_millis(timeout_ms);

        while start.elapsed() < timeout_duration {
            if condition() {
                return true;
            }
            sleep(Duration::from_millis(10)).await;
        }

        false
    }

    /// 等待异步条件满足
    pub async fn wait_for_async_condition<F, Fut>(condition: F, timeout_ms: u64) -> bool
    where
        F: Fn() -> Fut,
        Fut: std::future::Future<Output = bool>,
    {
        let start = std::time::Instant::now();
        let timeout_duration = Duration::from_millis(timeout_ms);

        while start.elapsed() < timeout_duration {
            if condition().await {
                return true;
            }
            sleep(Duration::from_millis(10)).await;
        }

        false
    }
}